package org.example;


import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousCloseException;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.util.Scanner;
import java.util.concurrent.*;

public class Main {


    private static final int extPort = 16000;
    private static final int clintPort = 16088;


    private static AsynchronousSocketChannel registerChannel;

    static BlockingQueue<AsynchronousSocketChannel> channelQueue = new LinkedBlockingQueue<>();

    public static void main(String[] args) throws IOException {

        final AsynchronousServerSocketChannel listener =
                AsynchronousServerSocketChannel.open().bind(new InetSocketAddress("localhost", clintPort));

        System.out.println("客户端提供连接服务启动，端口："+clintPort);
        listener.accept(null, new CompletionHandler<AsynchronousSocketChannel, Void>() {
            public void completed(AsynchronousSocketChannel ch, Void att) {

                // 接受连接，准备接收下一个连接
                listener.accept(null, this);

                // 处理连接
                clintHandle(ch);
            }

            public void failed(Throwable exc, Void att) {
                exc.printStackTrace();
            }
        });


        final AsynchronousServerSocketChannel extListener =
                AsynchronousServerSocketChannel.open().bind(new InetSocketAddress("localhost", extPort));

        System.out.println("外部访问的服务启动，端口："+extPort);

        extListener.accept(null, new CompletionHandler<AsynchronousSocketChannel, Void>() {

            private Future<Integer> writeFuture;

            public void completed(AsynchronousSocketChannel ch, Void att) {
                // 接受连接，准备接收下一个连接
                extListener.accept(null, this);

                try {
                    //判断是否有注册连接
                    if(registerChannel==null || !registerChannel.isOpen()){
                        try {
                            ch.close();
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                        return;
                    }
                    //下发指令告诉需要连接
                    ByteBuffer bf = ByteBuffer.wrap(new byte[]{1});
                    if(writeFuture != null){
                        writeFuture.get();
                    }
                    writeFuture = registerChannel.write(bf);

                    AsynchronousSocketChannel take = channelQueue.take();

                    //clint连接失败的
                    if(take == null){
                        ch.close();
                        return;
                    }

                    //交换数据
                    exchangeDataHandle(ch,take);

                } catch (Exception e) {
                    e.printStackTrace();
                }

            }

            public void failed(Throwable exc, Void att) {
                exc.printStackTrace();
            }
        });

        Scanner in = new Scanner(System.in);
        in.nextLine();


    }

    private static void clintHandle(AsynchronousSocketChannel ch) {

        final ByteBuffer buffer = ByteBuffer.allocate(1);
        ch.read(buffer, null, new CompletionHandler<Integer, Void>() {
            public void completed(Integer result, Void attachment) {
                buffer.flip();
                byte b = buffer.get();
                if (b == 0) {
                    registerChannel = ch;
                } else if(b == 1){
                    channelQueue.offer(ch);
                }else{
                    //clint连接不到
                    channelQueue.add(null);
                }

            }

            public void failed(Throwable exc, Void attachment) {
                exc.printStackTrace();
            }
        });
    }

    private static void exchangeDataHandle(AsynchronousSocketChannel ch1, AsynchronousSocketChannel ch2) {
        exchangeData(ch1,ch2);
        exchangeData(ch2,ch1);
    }

    private static void exchangeData(AsynchronousSocketChannel ch1, AsynchronousSocketChannel ch2) {
        try {
            final ByteBuffer buffer = ByteBuffer.allocate(1024);

            ch1.read(buffer, null, new CompletionHandler<Integer, CompletableFuture<Integer>>() {

                public void completed(Integer result, CompletableFuture<Integer> readAtt) {

                    CompletableFuture<Integer> future = new CompletableFuture<>();

                    if (result == -1 || buffer.position() == 0) {
                        // 处理连接关闭的情况或者没有数据可读的情况

                        try {
                            readAtt.get(3,TimeUnit.SECONDS);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

                        closeChannels(ch1, ch2);
                        return;
                    }

                    buffer.flip();

                    CompletionHandler readHandler = this;

                    ch2.write(buffer, future, new CompletionHandler<Integer, CompletableFuture<Integer>>() {
                        @Override
                        public void completed(Integer result, CompletableFuture<Integer> writeAtt) {

                            if (buffer.hasRemaining()) {
                                // 如果未完全写入，则继续写入
                                ch2.write(buffer, writeAtt, this);

                            } else {
                                writeAtt.complete(1);
                                // 清空buffer并继续读取
                                buffer.clear();
                                if(ch1.isOpen()){
                                    ch1.read(buffer, writeAtt, readHandler);
                                }
                            }

                        }

                        @Override
                        public void failed(Throwable exc, CompletableFuture<Integer> attachment) {
                            if(!(exc instanceof AsynchronousCloseException)){
                                exc.printStackTrace();
                            }
                            closeChannels(ch1, ch2);
                        }
                    });

                }

                public void failed(Throwable exc, CompletableFuture<Integer>  attachment) {
                    if(!(exc instanceof AsynchronousCloseException)){
                        exc.printStackTrace();
                    }
                    closeChannels(ch1, ch2);
                }
            });

        } catch (Exception ex) {
            ex.printStackTrace();
            closeChannels(ch1, ch2);
        }

    }

    private static void closeChannels(AsynchronousSocketChannel ch1, AsynchronousSocketChannel ch2) {
        if (ch1 != null && ch1.isOpen()) {
            try {
                ch1.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        if (ch2 != null && ch2.isOpen()) {
            try {
                ch2.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

}